Java 使用 Zookeeper 实现分布式锁
分布式锁设计思路
分布式锁有多种实现方式，比如通过数据库、redis 都可实现。作为分布式协同工具 ZooKeeper，当然也有着标准的实现方式。下面介绍在 Zookeeper 中如何实现排他锁。
- 每个客户端往
/Locks下创建临时有序节点/Locks/Lock_，创建成功后/Locks下面会有每个客户端对应的节点， 如/Locks/Lock_00000001 - 客户端取得
/Locks下子节点，并进行排序，判断排在最前面的是否为自己，如果自己的锁节点在第一位，代表获取锁成功 - 如果自己的锁节点不在第一位，则监听自己前一位的锁节点。例如，自己锁节点为
Lock_00000002，那么则监听Lock_00000001 - 当前一位锁节点
Lock_0000001对应的客户端执行完成，释放了锁，将会触发监听客户端Lock_00000002的逻辑 - 监听客户端重新执行第2步逻辑，判断自己是否获得了锁
基础连接
因为这个连接方式都大同小异，所以这里直接写在这里了，下面就不再重复这块代码
public class MyLock {
// zk连接对象
public static ZooKeeper zooKeeper;
// 计数器对象
private static CountDownLatch countDownLatch = new CountDownLatch(1);
// zk的连接串
private static String ip = "127.0.0.1:2181";
// 锁的节点名称
public static final String LOCK_ROOT_PATH = "/Locks";
public static final String LOCK_NODE_NAME = LOCK_ROOT_PATH + "/lock_";
private String lockPath;
private Watcher watcher = new Watcher() {
@Override
public void process(WatchedEvent event) {
if (event.getType() == Watcher.Event.EventType.NodeDeleted) {
synchronized (this) {
notifyAll();
}
}
}
};
/**
* 构造方法
*/
public MyLock() {
try {
// 打开zookeeper连接
zooKeeper = new ZooKeeper(ip, 5000, new Watcher() {
@Override
public void process(WatchedEvent event) {
if (event.getType() == Event.EventType.None) {
if (event.getState() == Event.KeeperState.SyncConnected) {
System.out.println("连接成功");
countDownLatch.countDown();
}
}
}
});
countDownLatch.await();
} catch (Exception e) {
e.printStackTrace();
}
}
}
创建节点
/**
* 创建锁节点
*/
private void createLock() {
try {
Stat exists = zooKeeper.exists(LOCK_ROOT_PATH, false);
// 判断/Locks是否存在，不存在则创建
if (exists == null) {
zooKeeper.create(LOCK_ROOT_PATH, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
// 创建临时有序节点，把这个保存到全局变量里面，下面还要用到
lockPath = zooKeeper.create(LOCK_NODE_NAME, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println("节点创建成功：" + lockPath);
} catch (Exception e) {
e.printStackTrace();
}
}
尝试获取锁
/**
* 尝试获取锁
*/
private void attemptLock() {
try {
// 获取 /Locks 节点下的所有子节点
List<String> children = zooKeeper.getChildren(LOCK_ROOT_PATH, false);
// 对子节点进行排序
Collections.sort(children);
// 获取上面创建的临时有序路径，例如这里的切割 /Locks/Lock_0000002 为 0000002（这里切割的目的是减少字符匹配次数），然后去查找 ZK 里面的的全部 Node
int index = children.indexOf(lockPath.substring((LOCK_NODE_NAME.length() - LOCK_ROOT_PATH.length()) + 1));
System.out.println(index);
// 如果创建的这个节点就是第一位，则表示取得了锁
if (index == 0) {
System.out.println("获取锁成功");
return;
} else {
// 如果没有找到，说明当前有服务在占用这个锁，则监听这个第一个 Node（/Locks/Lock_0000001）
String path = children.get(index - 1);
Stat exists = zooKeeper.exists(LOCK_ROOT_PATH + "/" + path, watcher);
// 如果 exists 为空表示占用锁的那个应用已经释放锁了，再次尝试获取锁
if (exists == null) {
attemptLock();
} else {
// 这里开始使用上面创建的 watcher 监听 NodeDeleted 事件
synchronized (watcher) {
// 释放自己获取的锁，直到被唤醒后才重新获取这个锁
watcher.wait();
}
attemptLock();
}
}
} catch (KeeperException | InterruptedException e) {
e.printStackTrace();
}
}
释放锁
这个没啥好说，就是用完把这个 Node 删除，标识释放了锁
/**
* 释放锁
*/
private void releaseLock() {
try {
// 删除临时有序节点
zooKeeper.delete(this.lockPath, -1);
zooKeeper.close();
System.out.println("锁已释放：" + lockPath);
} catch (InterruptedException | KeeperException e) {
e.printStackTrace();
}
}
获取锁
把上面的两个方法封装一下
/**
* 获取锁
*/
public void acquireLock() {
//创建锁节点
createLock();
//尝试获取锁
attemptLock();
}
编写测试类
/**
* 测试
*/
public static void main(String[] args) {
MyLock myLock = new MyLock();
// 获取锁
myLock.acquireLock();
sall();
// 释放锁
myLock.releaseLock();
}
public static void sall() {
System.out.println("售票开始");
// 线程休眠，模拟现实中耗时的操作
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("售票结束");
}
开三个窗口
可以观察到，它们会按序争抢这个锁
